package cn.kennylee.codehub.mq.lip.integration.kafka;

import cn.kennylee.codehub.mq.lip.integration.IntegrationMqConfig;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;

/**
 * <p> Kafka集成配置 </p>
 * <p>Created on 2025/6/12.</p>
 *
 * @author kennylee
 * @since 0.0.1
 */
@AutoConfiguration
@AutoConfigureAfter({IntegrationMqConfig.class})
@ConditionalOnClass(ConsumerFactory.class)
@ConditionalOnProperty(prefix = "app.mq.integration.kafka", name = "enable", havingValue = "true", matchIfMissing = true)
@Slf4j
@EnableKafka
public class IntegrationKafkaConfig {

    @Bean
    @ConditionalOnMissingBean
    @ConditionalOnProperty(prefix = "app.mq.integration.consumer", name = "enable", havingValue = "true", matchIfMissing = true)
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaContainerFactory(
        ConsumerFactory<String, String> consumerFactory) {

        ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // 并发数
        factory.setConcurrency(3);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
        return factory;
    }

    @Bean
    @ConditionalOnMissingBean
    @ConditionalOnProperty(prefix = "app.mq.integration.consumer", name = "enable", havingValue = "true", matchIfMissing = true)
    public KafkaMqListenerRegistrar kafkaMqListenerRegistrar(ConsumerFactory<String, String> consumerFactory) {
        return new KafkaMqListenerRegistrar(consumerFactory);
    }

    @PostConstruct
    public void init() {
        log.debug("Kafka集成配置，已加载");
    }
}
